// Copyright (c) 2023 Presto Labs Pte. Ltd.
// Author: chunhui

#include <csignal>
#include <unordered_map>
#include <vector>

#include <cxxopts.hpp>
#include <glog/logging.h>

#include "cc/appcoin2/support/feed/interval_feed_tool/util.h"
#include "coin2/app/driver.h"
#include "coin2/base/config/app_config.h"
#include "coin2/base/config/args.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/feed/timer/deadline_timer.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/base/order/order.h"
#include "coin2/exchange/base/symbology/product_encyclopedia.h"
#include "coin2/strategy/basis_strat/support/csv_writer.h"
#include "coin2/strategy/strategy.h"

namespace coin2::live_kline {

using namespace coin2::app;
using namespace coin2::base::config;
using coin2::exchange::base::feed::FeedUpdate;
using ::coin2::exchange::base::feed::timer::DeadlineTimer;
using ::coin2::exchange::base::feed::timer::FeedTimer;
using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::base::symbology::IProductInfo;
using coin2::exchange::base::symbology::ProductEncyclopedia;
using coin2::strategy::basis_strat::support::CsvWriter;
namespace fs = std::experimental::filesystem;

class GeneratingLiveKlineSubStrategy : public ::coin2::strategy::IStrategy {
 public:
  explicit GeneratingLiveKlineSubStrategy(
      int64_t ts,
      int64_t interval,
      const std::string& symbol_string,
      const std::string& file_path_root,
      ProductEncyclopedia* product_cache)
      : interval_(interval), product_cache_(product_cache), file_path_root_(file_path_root) {
    product_ = CreateProductFromUniqueString(symbol_string, ts);

    column_names_ = {
        "kline_timestamp",
        "open",
        "high",
        "low",
        "close",
        "bopen",
        "bhigh",
        "blow",
        "bclose",
        "volume",
        "buy_volume",
        "sell_volume",
        "turnover",
        "buy_turnover",
        "sell_turnover",
        "open_interest"};

    UpdateCsvPath(ts);
    book_ohlc_.resize(4, NAN);
    trade_ohlc_.resize(4, NAN);
    base_volume_.resize(3, 0);
    quote_volume_.resize(3, 0);
    pi_ = &(product_cache_->holder(*product_).product_info());
    if (std::isfinite(pi_->contract_value())) {
      contract_value_ = pi_->contract_value();
    }
  }

  void onBookFeed(const FeedUpdate& upd) {
    if (upd.book().Bid0().has_value() && upd.book().Ask0().has_value()) {
      double bid0 = upd.book().Bid0().value().price;
      double ask0 = upd.book().Ask0().value().price;
      double mid = 0.5 * (bid0 + ask0);
      if (book_reset_) {
        book_ohlc_[0] = mid;
        book_ohlc_[1] = mid;
        book_ohlc_[2] = mid;
        book_ohlc_[3] = mid;
        book_reset_ = false;
      } else {
        book_ohlc_[1] = std::max(mid, book_ohlc_[1]);
        book_ohlc_[2] = std::min(mid, book_ohlc_[2]);
        book_ohlc_[3] = mid;
      }
    }
  }

  void onTradeFeed(const FeedUpdate& upd) {
    double price = upd.trade().price;
    double qty = upd.trade().fill_qty;
    double dollar_volume = price * qty * contract_value_;
    if (upd.trade().side == coin2::exchange::feed::TradeSide::TRADE_BUY_SIDE) {
      quote_volume_[1] += dollar_volume;
      base_volume_[1] += qty;
      quote_volume_[0] += dollar_volume;
      base_volume_[0] += qty;
    } else if (upd.trade().side == coin2::exchange::feed::TradeSide::TRADE_SELL_SIDE) {
      quote_volume_[2] += dollar_volume;
      base_volume_[2] += qty;
      quote_volume_[0] += dollar_volume;
      base_volume_[0] += qty;
    }
    bool valid_trade = upd.trade().side == coin2::exchange::feed::TradeSide::TRADE_BUY_SIDE ||
                       upd.trade().side == coin2::exchange::feed::TradeSide::TRADE_SELL_SIDE;
    if (valid_trade) {
      if (trade_reset_) {
        trade_ohlc_[0] = price;
        trade_ohlc_[1] = price;
        trade_ohlc_[2] = price;
        trade_ohlc_[3] = price;
        trade_reset_ = false;
      } else {
        trade_ohlc_[1] = std::max(price, trade_ohlc_[1]);
        trade_ohlc_[2] = std::min(price, trade_ohlc_[2]);
        trade_ohlc_[3] = price;
      }
    }
  }

  void onOpenInterestFeed(const FeedUpdate& upd) {
    open_interest_ = upd.open_interest().open_interest_qty;
  }

  void OnHeartbeatFeed(int64_t ts, bool reset_only, bool suppress_create_new_file) {
    if (!reset_only) {
      WriteKline();
      if (!suppress_create_new_file) {
        UpdateCsvPath(ts);
      }
    }
    Reset(ts);
  }

  void UpdateCsvPath(int64_t ts) {
    if (ts / csv_file_change_period_ > last_csv_create_time_) {
      fs::path file_path = file_path_root_;
      std::string date_str = YmdFromTimestamp(ts);
      file_path /= date_str;
      if (!fs::is_directory(file_path)) {
        LOG(INFO) << "Try to create directory " << file_path;
        CHECK(fs::create_directories(file_path)) << "Fail to create directory " << file_path;
      }
      int64_t rounded_ts = ts / interval_ * interval_;
      std::string file_name = fmt::format("{}_{}.csv", product_->key_string(), rounded_ts);
      file_path /= file_name;
      csv_writer_ = std::make_unique<CsvWriter>(file_path.string(), column_names_);
      last_csv_create_time_ = ts / csv_file_change_period_;
    }
  }

  void WriteKline() {
    csv_writer_->AddItem<int64_t>(kline_start_ts_);
    csv_writer_->AddItems<double>(trade_ohlc_);
    csv_writer_->AddItems<double>(book_ohlc_);
    csv_writer_->AddItems<double>(base_volume_);
    csv_writer_->AddItems<double>(quote_volume_);
    csv_writer_->AddItem<double>(open_interest_);
    csv_writer_->WriteItems();
    csv_writer_->Flush();
  }

  void Reset(int64_t ts) {
    double close_mid = book_ohlc_[3];
    double close_trade = trade_ohlc_[3];
    std::fill(book_ohlc_.begin(), book_ohlc_.end(), close_mid);
    std::fill(trade_ohlc_.begin(), trade_ohlc_.end(), close_trade);
    std::fill(quote_volume_.begin(), quote_volume_.end(), 0);
    std::fill(base_volume_.begin(), base_volume_.end(), 0);
    book_reset_ = true;
    trade_reset_ = true;
    kline_start_ts_ = ts;
  }

 private:
  double contract_value_{1.0};
  bool book_reset_{true};
  bool trade_reset_{true};
  bool initialized_{false};
  int64_t kline_start_ts_{0};
  int64_t interval_;

  std::unique_ptr<IProduct> product_;
  std::vector<double> book_ohlc_;
  std::vector<double> trade_ohlc_;
  std::vector<double> quote_volume_;
  std::vector<double> base_volume_;  // Number of contract
  double open_interest_{NAN};
  std::unique_ptr<CsvWriter> csv_writer_{nullptr};
  std::vector<std::string> column_names_;

  ProductEncyclopedia* product_cache_;
  const std::string file_path_root_;
  const IProductInfo* pi_;

  int64_t last_csv_create_time_{0};
  int64_t csv_file_change_period_{86400'000'000'000LL};
};

class GeneratingLiveKlineStrategy : public ::coin2::strategy::IStrategy {
 public:
  GeneratingLiveKlineStrategy(const cxxopts::ParseResult& param, const std::string& file_path_root)
      : file_path_root_(file_path_root) {
    interval_ = static_cast<int64_t>(param["resolution"].as<int>()) * 1'000'000'000LL;
    strat_start_ts_ = GetCurrentTimestamp();
    current_rollover_time_ = strat_start_ts_ / rollover_period_;
    current_interval_time_ = strat_start_ts_ / interval_;
  }

  void onBookFeed(const FeedUpdate& upd) override {
    if (UpdateInterval(upd.timestamp())) {
      OnHeartbeatFeed(current_interval_time_ * interval_);
    }
    StopWhenRollover(upd.timestamp());
    const auto& key = upd.product().key_string();
    if (substrats_.count(key) == 0) {
      CreateSubStrat(key, upd.timestamp());
    }
    substrats_[key]->onBookFeed(upd);
  }

  void onTradeFeed(const FeedUpdate& upd) override {
    if (UpdateInterval(upd.timestamp())) {
      OnHeartbeatFeed(current_interval_time_ * interval_);
    }
    StopWhenRollover(upd.timestamp());
    const auto& key = upd.product().key_string();
    if (substrats_.count(key) == 0) {
      CreateSubStrat(key, upd.timestamp());
    }
    substrats_[key]->onTradeFeed(upd);
  }

  void onOpenInterestFeed(const FeedUpdate& upd) override {
    if (UpdateInterval(upd.timestamp())) {
      OnHeartbeatFeed(current_interval_time_ * interval_);
    }
    StopWhenRollover(upd.timestamp());
    const auto& key = upd.product().key_string();
    if (substrats_.count(key) == 0) {
      CreateSubStrat(key, upd.timestamp());
    }
    substrats_[key]->onOpenInterestFeed(upd);
  }

  void OnHeartbeatFeed(int64_t ts) {
    LOG(INFO) << "heartbeat " << ts;
    if (!initialized_) {
      heart_beat_count_++;
      if (heart_beat_count_ > 2) {
        initialized_ = true;
      }
    }
    bool suppress_create_new_file = IsReachingRolloverTime(ts);
    for (auto& substrat : usubstrats_) {
      substrat->OnHeartbeatFeed(ts, !initialized_, suppress_create_new_file);
    }
  }

  void CreateSubStrat(const std::string& symbol_key, int64_t ts) {
    usubstrats_.push_back(std::make_unique<GeneratingLiveKlineSubStrategy>(
        strat_start_ts_,
        interval_,
        symbol_key,
        file_path_root_,
        &product_cache_));
    substrats_[symbol_key] = usubstrats_.back().get();
  }

 private:
  bool IsReachingRolloverTime(int64_t ts) { return ts / rollover_period_ > current_rollover_time_; }
  void StopWhenRollover(int64_t ts) {
    if (IsReachingRolloverTime(ts)) {
      exit(1);
    }
  }

  bool UpdateInterval(int64_t ts) {
    if (ts / interval_ > current_interval_time_) {
      current_interval_time_ = ts / interval_;
      return true;
    }
    return false;
  }

  const std::string file_path_root_;
  int64_t interval_{60'000'000'000LL};
  int64_t current_interval_time_{0};
  int64_t strat_start_ts_{0};
  bool initialized_{false};
  int heart_beat_count_{0};
  std::vector<std::unique_ptr<GeneratingLiveKlineSubStrategy>> usubstrats_;
  std::unordered_map<std::string, GeneratingLiveKlineSubStrategy*> substrats_;
  ProductEncyclopedia product_cache_;

  int64_t current_rollover_time_{0};
  int64_t rollover_period_{86400'000'000'000LL};
};

std::string ResolvePaths(const cxxopts::ParseResult& param, const DriverConfig& app) {
  for (const auto& [key, exchange_config] : app.feed().exchanges()) {  // Assume only contains 1 mea
    (void)key;
    std::string resolution_str;
    interval_feed::GetResolutionStr(param["resolution"].as<int>(), &resolution_str);

    const std::string& mea = exchange_config.mea();
    std::string root_str = param["kline_root"].as<std::string>();
    CHECK_GT(root_str.size(), 0) << "empty kline root";
    fs::path filepath = root_str;
    CHECK(filepath.has_root_path()) << "File path should be absolute";
    filepath /= param["machine"].as<std::string>();
    filepath /= resolution_str;
    filepath /= mea;
    return filepath;
  }
  CHECK(false) << "empty driver";
  return "";
}

// void KillerHandler(int signum) {}

}  // namespace coin2::live_kline

int main(int argc, char* argv[]) {
  // signal(SIGTERM, coin2::live_kline::KillerHandler);
  coin2::app::DriverConfig app;
  cxxopts::Options opt("Live kline generator", "Generate kline");
  coin2::base::config::AddDriverOptions(&opt);

  opt.add_options()(
      "resolution",
      "length of interval in sec",
      cxxopts::value<int>()->default_value(
          "60"))("kline_root", "kline dump root dir", cxxopts::value<std::string>())(
      "machine",
      "used for directory creation",
      cxxopts::value<std::string>()->default_value("main"));

  auto param = opt.parse(argc, argv);
  CHECK(coin2::base::config::ParseDriverOptions(param, &app)) << "Fail to parse driver";
  ::coin2::base::InitLogging(argv[0], app.log());

  std::string file_path_root = coin2::live_kline::ResolvePaths(param, app);
  coin2::live_kline::GeneratingLiveKlineStrategy strategy(param, file_path_root);
  ::coin2::app::Driver driver(app, &strategy);
  driver.Run();

  return 0;
}
